package com.ruoyi.socket.server;


import com.alibaba.fastjson2.JSONObject;
import com.ruoyi.socket.data.DataService;
import com.ruoyi.socket.unit.ZipUtil;
import com.ruoyi.trade.domain.TPairs;
import com.ruoyi.vo.PairsListVo;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.*;
import io.netty.channel.group.ChannelGroup;
import io.netty.channel.group.DefaultChannelGroup;
import io.netty.handler.codec.http.*;
import io.netty.handler.codec.http.websocketx.*;
import io.netty.util.CharsetUtil;
import io.netty.util.ReferenceCountUtil;
import io.netty.util.concurrent.GlobalEventExecutor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.scheduling.annotation.EnableScheduling;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

import java.nio.channels.SocketChannel;
import java.util.ArrayList;
import java.util.List;
import java.util.Set;
import java.util.concurrent.*;

@Component
@EnableScheduling
@Slf4j
public class WebSocketServerHandler extends SimpleChannelInboundHandler<Object> {

	public static final ConcurrentHashMap<Channel, String> channel_map_kline = new ConcurrentHashMap<Channel, String>();
	public static final ConcurrentHashMap<Channel, String> channel_map_hour = new ConcurrentHashMap<Channel, String>();
	public static final ConcurrentHashMap<String, ChannelGroup> kline_group = new ConcurrentHashMap<String, ChannelGroup>();
	public static final ConcurrentHashMap<String, ChannelGroup> hour_group = new ConcurrentHashMap<String, ChannelGroup>();
	public static final String hour_group_key = "hourGroupKey";

	private DataService dataService;

	public WebSocketServerHandler(DataService dataService) {
		this.dataService=dataService;
	}

	private WebSocketServerHandshaker handshaker;

	//private static final Logger log = LoggerFactory.getLogger(NettyServerHandler.class);
	@Override
	protected void channelRead0(ChannelHandlerContext context, Object msg) throws Exception {
		if (msg instanceof FullHttpRequest) {
			handleHttpRequest(context, (FullHttpRequest) msg);
		} else if (msg instanceof WebSocketFrame) {
			handleWebSocketFrame(context, (WebSocketFrame) msg);
		}
	}
	@Override
	public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
		ctx.close();
	}

	private void handleHttpRequest(ChannelHandlerContext ctx, FullHttpRequest req) {
		if (!req.decoderResult().isSuccess()) {
			sendHttpResponse(ctx, req,
					new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.BAD_REQUEST));
			return;
		}
		if (req.method() != HttpMethod.GET) {
			sendHttpResponse(ctx, req, new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.FORBIDDEN));

			return;
		}
		WebSocketServerHandshakerFactory wsFactory = new WebSocketServerHandshakerFactory(getWebSocketLocation(req),
				null, true, 5242880);
		this.handshaker = wsFactory.newHandshaker(req);
		if (this.handshaker == null) {
			WebSocketServerHandshakerFactory.sendUnsupportedVersionResponse(ctx.channel());
		} else {
			this.handshaker.handshake(ctx.channel(), req);
		}
	}


	private void handleWebSocketFrame(ChannelHandlerContext ctx, WebSocketFrame frame) {
		if (frame instanceof CloseWebSocketFrame) {
			log.info("CloseWebSocketFrame");
			String mapKey=channel_map_kline.get(ctx.channel());
			if(mapKey!=null)
			{
				if(kline_group.containsKey(mapKey))
				{
					kline_group.get(mapKey).remove(ctx.channel());
					if(kline_group.get(mapKey).isEmpty())
					{
						kline_group.remove(mapKey);
					}
					channel_map_kline.remove(ctx.channel());
				}
				return;
			}
			String mapHourKey=channel_map_hour.get(ctx.channel());
			if(mapHourKey!=null)
			{
				if(hour_group.containsKey(mapHourKey))
				{
					hour_group.get(mapHourKey).remove(ctx.channel());
					if(hour_group.get(mapHourKey).isEmpty())
					{
						hour_group.remove(mapHourKey);
					}
					channel_map_hour.remove(ctx.channel());
				}
				return;
			}
			return;
		}
		if (frame instanceof io.netty.handler.codec.http.websocketx.PingWebSocketFrame) {
			log.info("PingWebSocketFrame");
			ctx.write(new PongWebSocketFrame(frame.content().retain()));
			return;
		}
		if (frame instanceof TextWebSocketFrame) {
			String twsf = ((TextWebSocketFrame) frame).text();
			log.info("TextWebSocketFrame:" + twsf);
			//第一次连接时查询K线数据
			if (twsf.contains("initKline")) {
				String[] params=twsf.split("-");
				String pairsName=params[1];
				String timeType=params[2];
				String type=params[3];//交易对类型（合约还是现货）
				Long start=Long.parseLong(params[4]);
				Long end=Long.parseLong(params[5]);
				String methodKey=pairsName+"-"+timeType;
				Set<String> set=dataService.getKlineByTime(pairsName, timeType,type,start, end);
				JSONObject jo=new JSONObject();
				jo.put("data", set);
				jo.put("method", "initKline");
				jo.put("pairs", pairsName);
				jo.put("timeType", timeType);
				ByteBuf heapBuffer = Unpooled.buffer(8);
				heapBuffer.writeBytes(ZipUtil.gzip(jo.toJSONString()).getBytes());
				ctx.channel().writeAndFlush(new BinaryWebSocketFrame(heapBuffer));

				ChannelGroup channelGroup = (ChannelGroup) kline_group.get(methodKey);
				if (channelGroup == null) {
					channelGroup = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);
				}
				//在Netty中，客户端连接是通过Channel对象来管理的。你可以在ChannelHandler中保存这些Channel的引用，以便在需要时向它们发送消息
				channelGroup.add(ctx.channel());
				kline_group.put(methodKey, channelGroup);
				//记录每一个客户端所请求的信息--如果存在则删除之前的组员信息
				if(channel_map_kline.containsKey(ctx.channel()))
				{
					String mapKey=channel_map_kline.get(ctx.channel());
					if(!mapKey.equals(methodKey))
					{
						channel_map_kline.put(ctx.channel(), methodKey);
						kline_group.get(mapKey).remove(ctx.channel());
						if(kline_group.get(mapKey).isEmpty())
						{
							kline_group.remove(mapKey);
						}
					}
				}else {
					channel_map_kline.put(ctx.channel(), methodKey);
				}
			}else if(twsf.equals("init24hour")){
				//第一次查询首页热门交易对数据
				List<TPairs> pairsList = dataService.getPairsList();
				if(pairsList != null && pairsList.size()>0){
					List<PairsListVo> list = new ArrayList<>();
					for(TPairs pairs:pairsList){
						String pairsName = pairs.getPairsName();
						list.add(dataService.getInint24Hour(pairsName));
					}
					JSONObject jo=new JSONObject();
					jo.put("data", list);
					jo.put("method", "init24hour");
					ByteBuf heapBuffer = Unpooled.buffer(8);
					heapBuffer.writeBytes(ZipUtil.gzip(jo.toJSONString()).getBytes());
					ctx.channel().writeAndFlush(new BinaryWebSocketFrame(heapBuffer));
					ChannelGroup channelGroup = (ChannelGroup) hour_group.get(hour_group_key);
					if (channelGroup == null) {
						channelGroup = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);
					}
					//在Netty中，客户端连接是通过Channel对象来管理的。你可以在ChannelHandler中保存这些Channel的引用，以便在需要时向它们发送消息
					channelGroup.add(ctx.channel());
					hour_group.put(hour_group_key, channelGroup);
					//记录每一个客户端所请求的信息--如果存在则删除之前的组员信息
					if(channel_map_hour.containsKey(ctx.channel()))
					{
						String mapKey=channel_map_hour.get(ctx.channel());
						if(!mapKey.equals(hour_group_key))
						{
							channel_map_hour.put(ctx.channel(), hour_group_key);
							hour_group.get(mapKey).remove(ctx.channel());
							if(hour_group.get(mapKey).isEmpty())
							{
								hour_group.remove(mapKey);
							}
						}
					}else {
						channel_map_hour.put(ctx.channel(), hour_group_key);
					}
				}

			}else if(twsf.equals("PING")){
				TextWebSocketFrame textWebSocketFrame = new TextWebSocketFrame("PONG");
				ctx.channel().writeAndFlush(textWebSocketFrame);
			}
		}

	}

	//给客户端发送k线数据（5秒执行一次）
	@Scheduled(fixedRate = 5000L)
	public void sendAllMessageKline() {
		ThreadPoolExecutor es = new ThreadPoolExecutor(50, 50, 0L, TimeUnit.SECONDS,
				new LinkedBlockingQueue<Runnable>(), new ThreadFactory() {
			@Override
			public Thread newThread(Runnable r) {
				Thread t = new Thread(r);
				t.setDaemon(true);
				return t;
			}
		});
		for (String methodKey : kline_group.keySet()) {
			es.submit(new sendKline(methodKey));
		}
		es.shutdown();
	}

	class sendKline implements Runnable{
		private String methodKey;

		sendKline(String methodKey){
			this.methodKey=methodKey;
		}
		@Override
		public void run() {
			try {
				String[] methodKeys=methodKey.split("-");
				JSONObject jo=new JSONObject();
				//获取最新的数据
				jo.put("data", dataService.getInintKline(methodKey));
				jo.put("method", "oneKline");
				jo.put("pairs", methodKeys[0]);
				jo.put("timeType", methodKeys[1]);
				ByteBuf heapBuffer = Unpooled.buffer(8);
				heapBuffer.writeBytes(ZipUtil.gzip(jo.toJSONString()).getBytes());
				kline_group.get(methodKey).writeAndFlush(new BinaryWebSocketFrame(heapBuffer));
			} catch (Exception e) {
				e.printStackTrace();
			}
		}

	}


//	//给客户端发送24小时实时数据（1秒执行一次）
//	@Scheduled(fixedRate = 1000L)
//	public void sendAllMessage24Hour() {
//		ThreadPoolExecutor es = new ThreadPoolExecutor(50, 50, 0L, TimeUnit.SECONDS,
//				new LinkedBlockingQueue<Runnable>(), new ThreadFactory() {
//			@Override
//			public Thread newThread(Runnable r) {
//				Thread t = new Thread(r);
//				t.setDaemon(true);
//				return t;
//			}
//		});
//		es.submit(new send24Hour());
//		es.shutdown();
//	}
//
//	class send24Hour implements Runnable{
//		@Override
//		public void run() {
//			try {
//				JSONObject jo=new JSONObject();
//				List<PairsListVo> list = new ArrayList<>();
//				List<TPairs> pairsList = dataService.getPairsList();
//				if(pairsList != null && pairsList.size()>0){
//					for(TPairs pairs:pairsList){
//						String pairsName = pairs.getPairsName();
//						list.add(dataService.getInint24Hour(pairsName));
//					}
//				}
//				//获取最新的数据
//				jo.put("data", list);
//				jo.put("method", "one24hour");
//				ByteBuf heapBuffer = Unpooled.buffer(8);
//				heapBuffer.writeBytes(ZipUtil.gzip(jo.toJSONString()).getBytes());
//				hour_group.get(hour_group_key).writeAndFlush(new BinaryWebSocketFrame(heapBuffer));
//			} catch (Exception e) {
//				e.printStackTrace();
//			}
//		}
//
//	}

	private static void sendHttpResponse(ChannelHandlerContext ctx, FullHttpRequest req, FullHttpResponse res) {
		if (res.status().code() != HttpResponseStatus.OK.code()) {
			ByteBuf buf = Unpooled.copiedBuffer(res.status().toString(), CharsetUtil.UTF_8);
			res.content().writeBytes(buf);
			buf.release();
			HttpUtil.setContentLength(res, res.content().readableBytes());
		}

		ChannelFuture f = ctx.channel().writeAndFlush(res);
		if (!HttpUtil.isKeepAlive(req) || res.status().code() != HttpResponseStatus.OK.code()) {
			f.addListener(ChannelFutureListener.CLOSE);
		}
	}

	private static String getWebSocketLocation(FullHttpRequest req) {
		String location = String.valueOf(req.headers().get(HttpHeaderNames.HOST)) + "/ws/websocket";
		log.info("location="+location);
		return "ws://" + location;
	}
}

